package com.xiaoyilin.fast;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

/**
 * KafkaProducer线程安全并且可以池化
 * <p>
 * 配置参数参见org.apache.kafka.clients.producer.ProducerConfig
 * <p>
 * send方法本身是异步
 * 同步发送可靠性高性能低
 * 发送消息三种模式 同步/异步/发后即忘
 * <p>
 * 同步
 * public    Future<RecordMetadata>  send(ProducerRecord<K ,  V>  record)
 * <p>
 * 异步
 * public    Future<RecordMetadata>  send(ProducerRecord<K ,  V>  record ,Callback  callback)
 * <p>
 * KafkaProducer会产生 重试异常/不可重试异常
 * <p>
 * 保证callbackl先于callback2之前调用
 * producer.send(recordl ,callbackl) ;
 * producer.send( record2 ,callback2);
 *
 * @author Administrator
 */
public class ProducerFastStart {

    public static final String topic = "test";
    /**
     * 不需要配置所有broker地址，客户端会自动发现
     */

    public static final String brokerList = "localhost:9092";

    public static void main(String[] args) {
        //使用Properties定义配置属性
        Properties properties = new Properties();
        /**
         * 生产者通过序列化器把对象转为字节数组通过网络发送到Kafka
         */
        //设置序列化key程序类
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        // 设置序列化value程序类
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        //设置生产者Broker服务器连接地址
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);
        properties.put(ProducerConfig.ACKS_CONFIG, "1");
        //重试次数-可重试异常才会重试
        properties.put(ProducerConfig.RETRIES_CONFIG, 10);
        //定义消息生产者对象，依靠此对象可以进行消息的传递
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);
        /*KafkaProducer<String,  String>  producer=  new  KafkaProducer<>(properties,
                new  StringSerializer() , new StringSerializer());*/
        ProducerRecord<String, String> producerRecord = new ProducerRecord<>("test", "hello");
        kafkaProducer.send(producerRecord);
        /**
         * 异步方式发送消息，可以指定一个回调函数，服务器返回响应时会调用该函数。
         * 我们可以在该函数中对一些异常信息进行处理，比如记录错误日志或者把消息写入错误消息文件以便日后分析。
         */
        kafkaProducer.send(producerRecord, new Callback() {
            @Override
            public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                /**
                 * 回调函数也保证分区有序
                 *
                 *  RecordMetadata和Exception互斥
                 *
                 *  发送成功recordMetadata不为null
                 *  发送失败e不为null
                 */
            }
        });
        //发完即忘只负责发送消息，不知道是否发送成功
        //kafkaProducer.send(producerRecord)
        /**
         * 同步发送会返回一个Future对象
         * 调用get()方法等待，就可以知道消息是否发送成功。
         * 如果服务器返回错误，get()方法会抛出异常。
         * 如果没有发生错误我们会得到一个RecordMetadata对象，可以用它获取消息的偏移量。
         */
        // kafkaProducer.send(producerRecord).get() ;
        //关闭资源-阻塞等待所有的发送请求发送完成后关闭资源
        kafkaProducer.close();
        //带超时关闭资源-到了时间强行关闭
        //kafkaProducer.close(Duration.ofMillis(10));
    }


}
